package com.gome.boot.config.mq.rocketmq;

import com.gome.boot.common.util.CommonUtils;
import com.gome.boot.common.util.PackageScanUtils;
import com.gome.boot.common.util.StringToUtils;
import com.gome.boot.config.mq.annotation.MQConsumer;
import com.gome.boot.config.mq.annotation.MQProducer;
import com.gome.boot.config.mq.annotation.RocketMQ;
import com.gome.rocketmq.client.extension.MessageConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * RocketMq注解扫描工具.
 * <br/>扫描MQConsumer和MQProducer注解.
 * @see MQConsumer
 * @see MQProducer
 *
 * @author baoxiufeng
 */
public class AnnotationBean implements DisposableBean, BeanFactoryPostProcessor, BeanPostProcessor, ApplicationContextAware {
    private static final Logger logger = LoggerFactory.getLogger(AnnotationBean.class);
    private String annotationPackage;
    private String[] annotationPackages;
    private ApplicationContext applicationContext;
    private final ConcurrentMap<Class<MessageConsumer>, MessagePushConsumerExtImpl> mmessagePushConsumers = new ConcurrentHashMap();
    private final ConcurrentMap<String, MessageProducerExtImpl> messageProducers = new ConcurrentHashMap();

    @Override
    public void destroy() throws Exception {
        Iterator<MessagePushConsumerExtImpl> pushConsumerIter = mmessagePushConsumers.values().iterator();
        while (pushConsumerIter.hasNext()) {
            applicationContext.getAutowireCapableBeanFactory().destroyBean(pushConsumerIter.next());
        }
        Iterator<MessageProducerExtImpl> producerIter = messageProducers.values().iterator();
        while (producerIter.hasNext()) {
            applicationContext.getAutowireCapableBeanFactory().destroyBean(producerIter.next());
        }
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        if (CommonUtils.isNotEmpty(annotationPackage)) {
            if (beanFactory instanceof BeanDefinitionRegistry) {
                PackageScanUtils.scan((BeanDefinitionRegistry) beanFactory, RocketMQ.class, StringToUtils.stringToArray(this.annotationPackage));
            }
        }
    }

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        if (!isMatchPackage(bean)) {
            return bean;
        }
        Class clazz = bean.getClass();
        if (AopUtils.isAopProxy(bean)) {
            clazz = AopUtils.getTargetClass(bean);
        }
        for (Field field : clazz.getDeclaredFields()) {
            try {
                if (!field.isAccessible()) {
                    field.setAccessible(true);
                }
                registProducer(field);
            } catch (Exception e) {
                throw new BeanInitializationException("Failed to init rocketmq producer at field " + field.getName() + " in class " + bean.getClass().getName(), e);
            }
        }
        return bean;
    }

    private void registProducer(Field field) {
        MQProducer producer = field.getAnnotation(MQProducer.class);
        if (producer != null && field.getType().isAssignableFrom(MessageProducerExtImpl.class)) {
            MessageProducerExtImpl messageProducer = new MessageProducerExtImpl(producer.orderly());
            messageProducer.setTopic(getEnvValue(producer.topic(), producer.isConfigKey()));
            messageProducer.setTags(getEnvValue(producer.tag(), producer.isConfigKey()));
            messageProducer.setGroup(MQConstants.DIAMOND_GROUP_ROCKETMQ_SERVER);
            messageProducer.setDataId(MQConstants.DIAMOND_DATAID_ROCKETMQ_SERVER);
            messageProducer.setOrderly(producer.orderly());
            messageProducer.setSerializeType(producer.serializeType() == null ? SerializeType.FASTJSON : producer.serializeType());
            messageProducer.setLevel((producer.level() == null ? MQDeleyLevel.LEVEL_0 : producer.level()).getLevel());
            ConfigurableListableBeanFactory beanFactory = ((AnnotationConfigApplicationContext) applicationContext).getBeanFactory();
            beanFactory.registerSingleton(field.getName(), messageProducer);
            applicationContext.getAutowireCapableBeanFactory().initializeBean(messageProducer, field.getName());
            messageProducers.put(field.getName(), messageProducer);
        }
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (!isMatchPackage(bean)) {
            return bean;
        }
        Class clazz = bean.getClass();
        if (AopUtils.isAopProxy(bean)) {
            clazz = AopUtils.getTargetClass(bean);
        }
        MQConsumer consumer = (MQConsumer) clazz.getAnnotation(MQConsumer.class);
        if (consumer != null && bean instanceof MessageConsumer) {
            MessagePushConsumerExtImpl messagePushConsumer = new MessagePushConsumerExtImpl();
            messagePushConsumer.setConsumerGroup(getEnvValue(consumer.consumerGroup(), consumer.isConfigKey(), ""));
            messagePushConsumer.setTopic(getEnvValue(consumer.topic(), consumer.isConfigKey()));
            messagePushConsumer.setSubExpression(getEnvValue(consumer.tag(), consumer.isConfigKey(), MQConstants.DEFAULT_CONSUMER_SUB_EXPRESSION));
            messagePushConsumer.setGroup(consumer.group());
            messagePushConsumer.setDataId(consumer.dataId());
            messagePushConsumer.setOrderly(consumer.orderly());
            messagePushConsumer.setSerializeType(consumer.serializeType() == null ? SerializeType.FASTJSON : consumer.serializeType());
            List<MessageConsumer> messageConsumers = new ArrayList<>(1);
            messageConsumers.add((MessageConsumer) bean);
            messagePushConsumer.setMessageConsumers(messageConsumers);
            ConfigurableListableBeanFactory beanFactory = ((AnnotationConfigApplicationContext) applicationContext).getBeanFactory();
            beanFactory.registerSingleton(beanName + "_pushConsumer", messagePushConsumer);
            applicationContext.getAutowireCapableBeanFactory().initializeBean(messagePushConsumer, beanName + "_pushConsumer");
            mmessagePushConsumers.put(clazz, messagePushConsumer);
        }
        return bean;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    private String getEnvValue(String key, boolean isConfigKey) {
        if (isConfigKey) {
            return applicationContext.getEnvironment().getRequiredProperty(key);
        }
        return key;
    }

    private String getEnvValue(String key, boolean isConfigKey, String defaultValue) {
        if (isConfigKey) {
            return applicationContext.getEnvironment().getProperty(key, defaultValue);
        }
        return key;
    }

    /**
     * 设置扫描包路径.
     *
     * @param annotationPackage 扫描包路径
     */
    public void setPackage(String annotationPackage) {
        this.annotationPackage = annotationPackage;
        this.annotationPackages = StringToUtils.stringToArray(annotationPackage);
    }

    /**
     * 判定给定对象类是否在扫描包路径下.
     *
     * @param bean 待判定的对象实例
     * @return 判定结果（true-匹配扫描包路径、false-不匹配扫描包路径）
     */
    private boolean isMatchPackage(Object bean) {
        if (CommonUtils.isEmpty(annotationPackages)) {
            return true;
        }
        Class clazz = bean.getClass();
        if (AopUtils.isAopProxy(bean)) {
            clazz = AopUtils.getTargetClass(bean);
        }
        String beanClassName = clazz.getName();
        for (String pkg : annotationPackages) {
            if (beanClassName.startsWith(pkg)) {
                return true;
            }
        }
        return false;
    }
}
